package hive.User_Defined_Functions;

import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;
import org.apache.hadoop.io.LongWritable;

/**
 * @describe: 多进一出，两个数求sum
 */
public class MyUDAFSUM extends GenericUDAFEvaluator {
    //输入数据序列化
    private PrimitiveObjectInspector inputOI;
    //返回值
    private LongWritable result;
    private boolean warned;

    public MyUDAFSUM() {
        this.warned = false;
    }

    /**
     * 这个方法返回udaf的返回类型。这里定义返回类型为long
     */
    @Override
    public ObjectInspector init(GenericUDAFEvaluator.Mode m, ObjectInspector[] parameters) throws HiveException {
        assert (parameters.length == 1);
        //输入类型初始化
        super.init(m, parameters);
        result = new LongWritable(0L);
        inputOI = ((PrimitiveObjectInspector) parameters[0]);
        //输出
        return PrimitiveObjectInspectorFactory.writableLongObjectInspector;
    }

    /**
     * 创建新的聚合计算需要的内存，用来存储mapper，combiner，reducer运算过程中的相加总和。
     *
     * @return GenericUDAFEvaluator.AggregationBuffer 用来存储聚集过程期间的结果
     */
    @Override
    public GenericUDAFEvaluator.AggregationBuffer getNewAggregationBuffer() throws HiveException {
        SumLong result = new SumLong();
        reset(result);
        return result;
    }

    /**
     * 重置聚合结果，可以方便重复使用相同的聚合
     */
    @Override
    public void reset(GenericUDAFEvaluator.AggregationBuffer agg) throws HiveException {
        SumLong myagg = (SumLong) agg;
        myagg.empty = true;
        myagg.sum = 0L;
    }

    /**
     * 迭代遍历原始的输入数据
     */
    @Override
    public void iterate(GenericUDAFEvaluator.AggregationBuffer agg, Object[] parameters) throws HiveException {
        assert (parameters.length == 1);
        try {
            merge(agg, parameters[0]);
        } catch (NumberFormatException e) {
            if (!(this.warned)) {
                this.warned = true;
            }
        }
    }

    /**
     * 合并部分聚合结果
     * <p>
     * 说明：
     * PrimitiveObjectInspectorUtils.getLong(数据， 数据的数据类型): 获取对应数据类型的数值
     */
    @Override
    public void merge(GenericUDAFEvaluator.AggregationBuffer agg, Object partial) throws HiveException {
        if (partial != null) {
            SumLong myagg = (SumLong) agg;
            myagg.sum += PrimitiveObjectInspectorUtils.getLong(partial, inputOI);
            myagg.empty = false;
        }
    }

    /**
     * 获得部分聚合的结果
     */
    @Override
    public Object terminatePartial(GenericUDAFEvaluator.AggregationBuffer agg) throws HiveException {
        return terminate(agg);
    }

    /**
     * 获得最终的聚合结果
     */
    @Override
    public Object terminate(GenericUDAFEvaluator.AggregationBuffer agg) {
        SumLong myagg = (SumLong) agg;
        if (myagg.empty) {
            return null;
        }
        result.set(myagg.sum);
        return result;
    }

    /**
     * 存储sum值的类
     */
    @GenericUDAFEvaluator.AggregationType(estimable = true)
    static class SumLong extends GenericUDAFEvaluator.AbstractAggregationBuffer {
        boolean empty;
        long sum;

        //预计存储空间大小 12个字节
        public int estimate() {
            return 12;
        }
    }
}
